#pragma once

#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/net/TcpClient.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_logger.hpp"
#include "mq_consumer.hpp"
#include "mq_channel.hpp"
#include "mq_worker.hpp"

namespace bitmq
{
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;
        using MessagePtr = std::shared_ptr<google::protobuf::Message>;
        using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
        using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
        using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
        Connection(const std::string& ip, int port, AsyncWorker::ptr& worker)
            : _latch(1)
            , _client(worker->loopthread.startLoop(), muduo::net::InetAddress(ip, port), "Client")
            , _worker(worker)
            , _channel_manager(std::make_shared<ChannelManager>())
            , _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))
            , _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
        {
            //注册业务回调函数
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::consumeResponse, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::basicResponse, this, 
                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            //注册连接以及消息回调
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            //连接服务器
            _client.connect();
            _latch.wait();
        }

        //创建信道
        Channel::ptr openChannel()
        {
            Channel::ptr channel = _channel_manager->create(_conn, _codec);
            bool ret = channel->openChannel();
            if (ret == false)
            {
                DLOG("打开信道失败!");
                return Channel::ptr();
            }
            return channel;
        }

        //关闭信道
        void closeChannel(const Channel::ptr& channel)
        {
            channel->closeChannel();
            _channel_manager->remove(channel->getCid());
        }

    private:
        void consumeResponse(const muduo::net::TcpConnectionPtr& conn, const basicConsumeResponsePtr& resp, muduo::Timestamp)
        {
            //1.找到信道
            Channel::ptr channel = _channel_manager->get(resp->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到信道信息!");
                return;
            }
            //2.封装异步任务(消息处理)抛入线程池
            _worker->pool.push([channel, resp](){
                channel->consume(resp);
            });
        }

        void basicResponse(const muduo::net::TcpConnectionPtr& conn, const basicCommonResponsePtr& resp, muduo::Timestamp)
        {
            //1.找到信道
            Channel::ptr channel = _channel_manager->get(resp->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到信道信息!");
                return;
            }
            //2.将得到的响应对象添加到信道基础响应hash_map中
            channel->putBasicResponse(resp);
        }

        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if (conn->connected())
            {
                DLOG("连接建立成功");
                _latch.countDown();
                _conn = conn;
            }
            else 
            {
                DLOG("连接断开");
                _conn.reset();
            }
        }

        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp)
        {
            std::cout << "onUnknownMessage: " << message->GetTypeName() << std::endl;
            conn->shutdown();
        }

    private:
        muduo::CountDownLatch _latch; //实现同步
        muduo::net::TcpConnectionPtr _conn; //客户端与服务器的连接
        muduo::net::TcpClient _client;    //客户端
        ProtobufDispatcher _dispatcher; //请求分发器
        ProtobufCodecPtr _codec; //协议处理器
        AsyncWorker::ptr _worker;
        ChannelManager::ptr _channel_manager;
    };
}
